/*
 *  Copyright (c) 2014, Peter Haag
 *  Copyright (c) 2009, Peter Haag
 *  Copyright (c) 2004-2008, SWITCH - Teleinformatikdienste fuer Lehre und Forschung
 *  All rights reserved.
 *  
 *  Redistribution and use in source and binary forms, with or without 
 *  modification, are permitted provided that the following conditions are met:
 *  
 *   * Redistributions of source code must retain the above copyright notice, 
 *     this list of conditions and the following disclaimer.
 *   * Redistributions in binary form must reproduce the above copyright notice, 
 *     this list of conditions and the following disclaimer in the documentation 
 *     and/or other materials provided with the distribution.
 *   * Neither the name of the author nor the names of its contributors may be 
 *     used to endorse or promote products derived from this software without 
 *     specific prior written permission.
 *  
 *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
 *  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
 *  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 
 *  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 
 *  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 
 *  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 
 *  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 
 *  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 
 *  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 
 *  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 
 *  POSSIBILITY OF SUCH DAMAGE.
 *  
 *  $Author: haag $
 *
 *  $Id: netflow_v5_v7.c 69 2010-09-09 07:17:43Z haag $
 *
 *  $LastChangedRevision: 69 $
 *	
 */

#include <netinet/in.h>


#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <stdlib.h>
#include <syslog.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <errno.h>
#include <string.h>

#ifdef HAVE_STDINT_H
#include <stdint.h>
#endif

#include "util.h"
#include "nffile.h"
#include "nfx.h"
//#include "nfnet.h"
#include "nf_common.h"
#include "bookkeeper.h"
//#include "nfxstat.h"
#include "collector.h"
#include "exporter.h"
#include "netflow_v5_v7.h"






#ifndef DEVEL
#   define dbg_printf(...) /* printf(__VA_ARGS__) */
#else
#   define dbg_printf(...) printf(__VA_ARGS__)
#endif



extern int verbose;
extern extension_descriptor_t extension_descriptor[];
extern uint32_t default_sampling;
extern uint32_t overwrite_sampling;

/* module limited globals */
static extension_info_t v5_extension_info;		// common for all v5 records
static uint16_t  v5_output_record_base_size;

// All required extension to save full v5 records
static uint16_t v5_full_mapp[] = { EX_IO_SNMP_2, EX_AS_2, EX_MULIPLE, EX_NEXT_HOP_v4, EX_ROUTER_IP_v4, EX_ROUTER_ID, EX_RECEIVED, 0 };

// to simplify, assume blocks with 64 bit counters to check for enough buffer space
// regardless if 32 or 64 bit packet/byte counters
#define V5_BLOCK_DATA_SIZE (sizeof(ipv4_block_t) - sizeof(uint32_t) + 2 * sizeof(uint64_t))

typedef struct exporter_v5_s {
	// identical to generic_exporter_t
	struct exporter_v5_s *next;

	// generic exporter information
	exporter_info_record_t info;

	uint64_t	packets;			// number of packets sent by this exporter
	uint64_t	flows;				// number of flow records sent by this exporter
	uint32_t	sequence_failure;	// number of sequence failues

	// generic sampler
	generic_sampler_t		*sampler;
	// end of generic_exporter_t

	// sequence vars
	int64_t	 last_sequence;
	int64_t  sequence, distance;
	int64_t  last_count;

	int		first;

	// extension map
	extension_map_t 	 *extension_map;

} exporter_v5_t;

static inline exporter_v5_t *GetExporter(FlowSource_t *fs, netflow_v5_header_t *header);

/* functions */

//#include "nffile_inline.c"

int Init_v5_v7_input(void) {
int i, id, map_index;
int extension_size;
uint16_t	map_size;

	extension_size   = 0;
	// prepare v5 extension map
	v5_extension_info.map		   = NULL;
	v5_extension_info.next		   = NULL;
	v5_extension_info.offset_cache = NULL;
	v5_extension_info.ref_count	   = 0;

	// default map - 0 extensions
	map_size 		 = sizeof(extension_map_t);
	i=0;
	while ( (id = v5_full_mapp[i]) != 0  ) {
		if ( extension_descriptor[id].enabled ) {
			extension_size += extension_descriptor[id].size;
			map_size += sizeof(uint16_t);
		}
		i++;
	}
	// extension_size contains the sum of all optional extensions
	// caculate the record size without counters!
	v5_output_record_base_size = COMMON_RECORD_DATA_SIZE + 8 + extension_size;  // + 8 for 2 x IPv4 addr
 
	// align 32 bits
	if ( ( map_size & 0x3 ) != 0 )
		map_size += 2;

	// Create a generic v5 extension map
	v5_extension_info.map = (extension_map_t *)malloc((size_t)map_size);
	if ( !v5_extension_info.map ) {
		syslog(LOG_ERR, "Process_v5: malloc() error in %s line %d: %s\n", __FILE__, __LINE__, strerror (errno));
		return 0;
	}
	v5_extension_info.map->type 	  	  = ExtensionMapType;
	v5_extension_info.map->size 	  	  = map_size;
	v5_extension_info.map->map_id 	  	  = INIT_ID;		
	v5_extension_info.map->extension_size = extension_size;

	// see netflow_v5_v7.h for extension map description
	map_index = 0;
	i=0;
	while ( (id = v5_full_mapp[i]) != 0 ) {
		if ( extension_descriptor[id].enabled )
			v5_extension_info.map->ex_id[map_index++] = id;
		i++;
	}
	v5_extension_info.map->ex_id[map_index] = 0;

	return 1;
} // End of Init_v5_input

/*
 * functions used for receiving netflow v5 records
 */


static inline exporter_v5_t *GetExporter(FlowSource_t *fs, netflow_v5_header_t *header) {
exporter_v5_t **e = (exporter_v5_t **)&(fs->exporter_data);
generic_sampler_t *sampler;
uint16_t	engine_tag = ntohs(header->engine_tag);
uint16_t	version    = ntohs(header->version);
#define IP_STRING_LEN   40
char ipstr[IP_STRING_LEN];

	// search the appropriate exporter engine
	while ( *e ) {
		if ( (*e)->info.version == version && (*e)->info.id == engine_tag &&
			 (*e)->info.ip.v6[0] == fs->ip.v6[0] && (*e)->info.ip.v6[1] == fs->ip.v6[1]) 
			return *e;
		e = &((*e)->next);
	}

	// nothing found
	*e = (exporter_v5_t *)malloc(sizeof(exporter_v5_t));
	if ( !(*e)) {
		syslog(LOG_ERR, "Process_v5: malloc() error in %s line %d: %s\n", __FILE__, __LINE__, strerror (errno));
		return NULL;
	}
	memset((void *)(*e), 0, sizeof(exporter_v5_t));
	(*e)->next	 			= NULL;
	(*e)->info.header.type  = ExporterInfoRecordType;
	(*e)->info.header.size  = sizeof(exporter_info_record_t);
	(*e)->info.version 		= version;
	(*e)->info.id			= engine_tag;
	(*e)->info.ip			= fs->ip;
	(*e)->info.sa_family	= fs->sa_family;
	(*e)->sequence_failure	= 0;
	(*e)->packets			= 0;
	(*e)->flows				= 0;
	(*e)->first	 			= 1;

	sampler = (generic_sampler_t *)malloc(sizeof(generic_sampler_t));
	if ( !sampler ) {
		syslog(LOG_ERR, "Process_v5: malloc() error in %s line %d: %s\n", __FILE__, __LINE__, strerror (errno));
		return NULL;
	}
	(*e)->sampler = sampler;

	sampler->info.header.type 	= SamplerInfoRecordype;
	sampler->info.header.size	= sizeof(sampler_info_record_t);
	sampler->info.id			= -1;
	sampler->info.mode			= (0xC000 & ntohs(header->sampling_interval)) >> 14;
	sampler->info.interval		= 0x3fff & ntohs(header->sampling_interval);
	sampler->next				= NULL;

	// default is global default_sampling ( user defined or unsampled => 1 )
	if ( sampler->info.interval == 0 )
		sampler->info.interval = default_sampling;

	// copy the v5 generic extension map
	(*e)->extension_map		= (extension_map_t *)malloc(v5_extension_info.map->size);
	if ( !(*e)->extension_map ) {
		syslog(LOG_ERR, "Process_v5: malloc() error in %s line %d: %s\n", __FILE__, __LINE__, strerror (errno));
		free(*e);
		*e = NULL;
		return NULL;
	}
	memcpy((void *)(*e)->extension_map, (void *)v5_extension_info.map, v5_extension_info.map->size);

	if ( !AddExtensionMap(fs, (*e)->extension_map) ) {
		// bad - we must free this map and fail - otherwise data can not be read any more
		free((*e)->extension_map);
		free(*e);
		*e = NULL;
		return NULL;
	}

	(*e)->info.sysid = 0;
	//FlushInfoExporter(fs, &((*e)->info));
	sampler->info.exporter_sysid		= (*e)->info.sysid;
	//FlushInfoSampler(fs, &(sampler->info));

	if ( fs->sa_family == AF_INET ) {
		uint32_t _ip = htonl(fs->ip.v4);
		inet_ntop(AF_INET, &_ip, ipstr, sizeof(ipstr));
	} else if ( fs->sa_family == AF_INET6 ) {
		uint64_t _ip[2];
		_ip[0] = htonll(fs->ip.v6[0]);
		_ip[1] = htonll(fs->ip.v6[1]);
		inet_ntop(AF_INET6, &_ip, ipstr, sizeof(ipstr));
	} else {
		strncpy(ipstr, "<unknown>", IP_STRING_LEN);
	}



	dbg_printf("New Exporter: v5 SysID: %u, Extension ID: %i, IP: %s, Sampling Mode: %i, Sampling Interval: %u\n", 
		(*e)->info.sysid, (*e)->extension_map->map_id, ipstr, sampler->info.mode	,sampler->info.interval);
	syslog(LOG_INFO, "Process_v5: New exporter: SysID: %u, engine id %u, type %u, IP: %s, Sampling Mode: %i, Sampling Interval: %u\n", 
		(*e)->info.sysid, ( engine_tag & 0xFF ),( (engine_tag >> 8) & 0xFF ), ipstr, sampler->info.mode	,sampler->info.interval );

	if ( overwrite_sampling > 0 )  {
		sampler->info.interval = overwrite_sampling;
		syslog(LOG_INFO, "Process_v5: Hard overwrite sampling rate: %u\n", sampler->info.interval);
	}

	return (*e);

} // End of GetExporter

int Process_v5_v7(void *in_buff, ssize_t in_buff_cnt, FlowSource_t *fs,  zw_dp_xflow_stats_t *pstats) {
    netflow_v5_header_t	*v5_header;
    netflow_v5_record_t *v5_record;
    exporter_v5_t 		*exporter;
    uint16_t	count = 0;
    int			i, done, version, flow_record_length;
    ssize_t		size_left; 

#define ETH_P_IP        0x0800
	/*
	 * v7 is treated as v5. It differes only in the record length, for what we process.
	 */

	// map v5 data structure to input buffer
	v5_header 	= (netflow_v5_header_t *)in_buff;

	exporter = GetExporter(fs, v5_header);
	if ( !exporter ) {
		syslog(LOG_ERR,"Process_v5: Exporter NULL: Abort v5/v7 record processing");
		return 0;
	}
	exporter->packets++;

	version = ntohs(v5_header->version);
	flow_record_length = version == 5 ? NETFLOW_V5_RECORD_LENGTH : NETFLOW_V7_RECORD_LENGTH;

	// this many data to process
	size_left	= in_buff_cnt;

	done = 0;
	while ( !done ) {
		/* Process header */

		// count check
  		count	= ntohs(v5_header->count);
		if ( count > NETFLOW_V5_MAX_RECORDS ) {
			syslog(LOG_ERR,"Process_v5: Unexpected record count in header: %i. Abort v5/v7 record processing", count);
			return 0;
		}

		// input buffer size check for all expected records
		if ( size_left < ( NETFLOW_V5_HEADER_LENGTH + count * flow_record_length) ) {
			syslog(LOG_ERR,"Process_v5: Not enough data to process v5 record. Abort v5/v7 record processing");
			return 0;
		}

		// sequence check
		if ( exporter->first ) {
			exporter->last_sequence = ntohl(v5_header->flow_sequence);
			exporter->sequence 	  	= exporter->last_sequence;
			exporter->first 	  	= 0;
		} else {
			exporter->last_sequence = exporter->sequence;
			exporter->sequence 	  = ntohl(v5_header->flow_sequence);
			exporter->distance 	  = exporter->sequence - exporter->last_sequence;
			// handle overflow
			if (exporter->distance < 0) {
				exporter->distance = 0xffffffff + exporter->distance  +1;
			}
			if (exporter->distance != exporter->last_count) {
#define delta(a,b) ( (a)>(b) ? (a)-(b) : (b)-(a) )
				//fs->nffile->stat_record->sequence_failure++;
				exporter->sequence_failure++;
				/*
				syslog(LOG_ERR,"Flow v%d sequence last:%llu now:%llu mismatch. Missing: dist:%lu flows",
					version, exporter->last_sequence, exporter->sequence, exporter->distance);
				*/

			}
		}
		exporter->last_count  = count;

  		v5_header->SysUptime	 = ntohl(v5_header->SysUptime);
  		v5_header->unix_secs	 = ntohl(v5_header->unix_secs);
  		v5_header->unix_nsecs	 = ntohl(v5_header->unix_nsecs);

		// process all records
		v5_record	= (netflow_v5_record_t *)((pointer_addr_t)v5_header + NETFLOW_V5_HEADER_LENGTH);

		/* loop over each records associated with this header */
		for (i = 0; i < count; i++, pstats++) {  
            pstats->ethtype= ETH_P_IP;
  			pstats->tcp_flags	  = v5_record->tcp_flags;
  			pstats->prot			  = v5_record->prot;
  			pstats->dip= v5_record->dstaddr;
            pstats->sip= v5_record->srcaddr;
            pstats->sport = v5_record->srcport ;
            pstats->dport = v5_record->dstport ;

			pstats->pkts 	= (uint64_t)ntohl(v5_record->dPkts);
			pstats->Bytes= (uint64_t)ntohl(v5_record->dOctets);

			//计算周期（单位秒）
			pstats->sample_rate= (uint16_t)((ntohl(v5_record->Last) - ntohl(v5_record->First))/1000);

            //if(v5_record->Last >= v5_record->First)
            //    printf("last:%u ,first:%u during:%d\n", v5_record->Last, v5_record->First, v5_record->Last - v5_record->First);
			// advance to next input flow record
			v5_record		= (netflow_v5_record_t *)((pointer_addr_t)v5_record + flow_record_length);
		} // End of foreach v5 record

		// still to go for this many input bytes
		size_left 	-= NETFLOW_V5_HEADER_LENGTH + count * flow_record_length;

		// next header
		v5_header	= (netflow_v5_header_t *)v5_record;

		// should never be < 0
		done = size_left <= 0;

	} // End of while !done

	return count;

} /* End of Process_v5 */

